package com.xxx.redismq.consumer;

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import com.xxx.redismq.common.utils.JedisUtils;
import com.xxx.redismq.common.utils.RedisMqUtil;

import javax.annotation.Resource;
import java.util.Set;

/**
 * redis mq监听器，监听队列中的消息是否到期
 */
@Component
public class MqMonitor  {

    @Resource
    private JedisUtils jedisUtils;

    @Resource
    private RedisMqUtil redisMQ;

    /**
     * 消息队列监听器
     * 监听所有路由器，将消息队列中的消息路由到待消费列表
     * 每隔5秒执行一次
     */
    @Scheduled(cron="*/5 * * * * *")
    public void monitor() {
        // 获取消息路由
        int route_size;
        if (null == redisMQ.getRoutes() || (route_size = redisMQ.getRoutes().size())<1) {
        	return;
        }
        String queue, list;
        Set<String> set;
        for (int i = 0; i < route_size; i++) {
            queue = redisMQ.getRoutes().get(i).getQueue();
            list = redisMQ.getRoutes().get(i).getList();
            //获取队列中的消息，根据score降序，score大的优先，取出前N个元素放入到待消费列表里
            set = jedisUtils.getSoredSetByRange(queue, 0, redisMQ.getMonitorCount(), true);
            if (null != set) {
                long current = System.currentTimeMillis();
                long score;
                for (String id : set) {
                    //取出score（score为createTime+delay+priority）
                    score = jedisUtils.getScore(queue, id).longValue();
                    if (current >= score) { // 比较zset里的score和当前时间戳大小，如果当前时间戳>score，说明时间到了，可以消费
                        // 添加到待消费列表中（先进先出队列）
                        if (jedisUtils.insertList(list, id)) {
                            // 添加成功后再从redis的zset中删除该成员（消息ID）
                            redisMQ.delMsgFromMq(queue, id);
                        }
                    }
                }
            }
        }
    }
}
